package com.xueshan.disruptor.lmax.producer.impl;

import com.lmax.disruptor.RingBuffer;
import com.xueshan.disruptor.lmax.consumer.DisruptorConsumer;
import com.xueshan.disruptor.lmax.consumer.MyConsumer;
import com.xueshan.disruptor.lmax.event.LongEvent;
import com.xueshan.disruptor.lmax.event.ObjectEvent;
import com.xueshan.disruptor.lmax.producer.MyProducer;
import com.xueshan.disruptor.lmax.queue.DisruptorQueue;
import com.xueshan.disruptor.lmax.queue.DisruptorQueueFactory;
import com.xueshan.disruptor.lmax.queue.RingBufferFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

/**
 * created with IDEA
 *
 * @author:huqm
 * @date:2021/3/16
 * @time:18:01 <p>
 *
 * </p>
 */
@Service
@Slf4j
public class MyProducerImpl implements MyProducer {

    @Override
    public void calcHandleEventsQueue(long value) {
        MyConsumer[] mc = new MyConsumer[3];
        for(int i=0;i<mc.length;i++){
            mc[i]= new MyConsumer();
        }
        DisruptorQueue<LongEvent> workPoolQueue = DisruptorQueueFactory.getHandleEventsQueue(1024*1024, false, mc);
        log.info("生产者开始 {}",System.currentTimeMillis());
        for(int i= 0 ;i <1024*1024;i++) {
            workPoolQueue.add(new LongEvent());
        }
        log.info("生产者发布完成");
    }

    @Override
    public void calcPoint2Point(long value) {
        LongEvent event = new LongEvent();
        event.setLat1(11L);
        DisruptorConsumer<LongEvent>[] mc = new MyConsumer[3];
        for(int i=0;i<mc.length;i++){
            mc[i]= new MyConsumer();
        }
        RingBuffer<ObjectEvent> ringBuffer = RingBufferFactory.getWorkRingBuffer(1024 * 1024, false, mc);
        log.info("点对点发布开始");
        for(int i=0;i<6000;i++) {
            long sequence = ringBuffer.next(); // 获得下一个Event槽的下标
            try {
                // 给Event填充数据
                ObjectEvent<LongEvent> objectEvent = ringBuffer.get(sequence);
                objectEvent.setObj(event);

            } finally {
                // 发布Event，激活观察者去消费， 将sequence传递给该消费者
                // 注意，最后的 ringBuffer.publish() 方法必须包含在 finally 中以确保必须得到调用；如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。
                ringBuffer.publish(sequence);
            }
        }
        log.info("点对点发布完成");
    }
}
